package com.zjh.storm;

import com.zjh.storm.bolts.WordCounterBolt;
import com.zjh.storm.bolts.WordNormalizerBolt;
import com.zjh.storm.spout.WordReaderSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * Created by zhoujianhua on 2017/9/6.
 */
public class TopologMain {
    public static void main(String[] args) throws InterruptedException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-reader", new WordReaderSpout());
        builder.setBolt("word-normalizer", new WordNormalizerBolt()).shuffleGrouping("word-reader");
        builder.setBolt("word-counter", new WordCounterBolt()).fieldsGrouping("word-normalizer", new Fields("word"));

        Config config = new Config();
        config.put("wordsFile", args[0]);
        //config.setDebug(true);

        config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-ounter-topology", config, builder.createTopology());

        Thread.sleep(5000);
        cluster.shutdown();

    }
}
